package com.bigdata.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class WordCount {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //    1.从本地加载配置文件得到配置对象
        Configuration conf = new Configuration(true);


        // 在window下并且是local本地模式执行mr任务，需要加一些配置
        // windows异构平台运行
        conf.set("mapreduce.app-submission.cross-platform", "true");
        //决定了集群运行
        conf.set("mapreduce.framework.name", "local");


//    2.使用配置对象创建Job
        Job job = Job.getInstance(conf);
        // job.setJar("D:\\bigdata_project\\bigdata_2307\\bigdata_hadoop\\target\\bigdata_hadoop-1.0-SNAPSHOT.jar");


//    3.设置任务的主启动类，job名称
        job.setJarByClass(WordCount.class);
        job.setJobName("WordCount");

        // 参数个性化
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] argsArr = parser.getRemainingArgs();

        /*
        4.设置HDFS输入、输出路径
        输入路径可以设置多个，会把多个路径下的所有的文件都参与MR计算
        输出路径只能设置一个，并且目录必须要为空，如果存在代码会报错，怕我们覆盖了原来的数据
         */
        Path inputPath = new Path(argsArr[0]);
        Path outputPath = new Path(argsArr[1]);
        TextInputFormat.addInputPath(job, inputPath);

        // 判断目录是否存在，如果存在的话，把目录删除
        if (outputPath.getFileSystem(conf).exists(outputPath)) {
            outputPath.getFileSystem(conf).delete(outputPath, true);
        }

        TextOutputFormat.setOutputPath(job, outputPath);

//    5.设置Mapper类、Mapper的输出key、value类型
        job.setMapperClass(WordMapper.class);
        // key需要具备比较的功能
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

//    6.设置排序比较器、分区器、分组比较器 后面案例再写

//    7.设置reducer类，reduce输出类型（可选）
        job.setReducerClass(WordReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置自定义的排序比较器
        // job.setSortComparatorClass();
        // 设置分组比较器
        // job.setGroupingComparatorClass();

        /*
         combine, 内存数据排序之后溢写之前，溢写的io变少
         调优点，可以优化MR程序的执行速度
         */
        job.setCombinerClass(WordCombiner.class);


        // 默认reduce task的数量是1个,可以自定义的指定
        // job.setNumReduceTasks(2);

//    8.提交任务, true 表示监控任务
        job.waitForCompletion(true);


    }


}
